package com.shujia.dwi

import com.shujia.grid.Grid
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

import java.awt.geom.Point2D
import java.text.SimpleDateFormat
import scala.collection.mutable.ListBuffer

object StayPointApp extends Logging {
  def main(args: Array[String]): Unit = {

    /**
     * 停留表处理逻辑
     * 将同一个人在同一个网格中的多条数据合并成一条
     *
     */

    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id: String = args(0)

    log.info(s"当前时间分区为：$day_id")

    val spark: SparkSession = SparkSession
      .builder()
      .appName("StayPointApp")
      .config("spark.sql.shuffle.partitions", "10")
      .enableHiveSupport() //使用hive元数据
      .getOrCreate()

    import spark.implicits._

    /**
     * mdn string comment '手机号码'
     * ,start_time string comment '业务时间'
     * ,county_id string comment '区县编码'
     * ,longi string comment '经度'
     * ,lati string comment '纬度'
     * ,bsid string comment '基站标识'
     * ,grid_id string comment '网格号'
     * ,biz_type string comment '业务类型'
     * ,event_type string comment '事件类型'
     * ,data_source string comment '数据源'
     *
     */

    //1、读取位置融合表数据
    val mergeLocation: DataFrame = spark.sql(s"select mdn,start_time,grid_id,county_id from dwi.dwi_res_regn_mergelocation_msk_d where day_id=$day_id")

    //2、将数据转换成kv格式，key是手机号，value是网格号、区域编码、停留时间
    val kvRDD: RDD[(String, (String, String, String, String))] = mergeLocation
      .rdd
      .map {
        case Row(mdn: String, start_time: String, grid_id: String, county_id: String) =>

          //取出一个网格号的开始时间和结束时间
          val endTime: String = start_time.split(",")(0)
          val startTime: String = start_time.split(",")(1)

          (mdn, (endTime, startTime, grid_id, county_id))

      }

    //3、按照手机号key进行分组
    val groupRDD: RDD[(String, Iterable[(String, String, String, String)])] = kvRDD.groupByKey()

    val pointRDD: RDD[(String, String, String, String, String, String, String, String)] = groupRDD.flatMap {

      case (mdn: String, points: Iterable[(String, String, String, String)]) =>

        //对一个人所有的点按照时间进行升序排序
        val sortPoint: List[(String, String, String, String)] = points.toList.sortBy(_._1)

        //最终合并的结果
        val resultPoint: ListBuffer[(String, String, String, String)] = new ListBuffer[(String, String, String, String)]

        //第一个点的结束时间
        var pointEndTime: String = points.head._1
        //第一个点的开始时间
        var pointStartTime: String = points.head._2
        //第一个网格号
        var headGrid: String = points.head._3
        //最后一个区域编号
        var lastCountyId: String = points.last._4

        //遍历一个人所有的点
        sortPoint.foreach {

          case (endTime: String, startTime: String, grid_id: String, county_id: String) =>

            //如果当前网格和上一个网格是一样的，就是同一个组
            if (grid_id.equals(headGrid)) {

              headGrid = grid_id
              //更新结束时间
              pointEndTime = endTime
              //更新区域编号
              lastCountyId = county_id

            } else {

              //将上一个组保存起来
              resultPoint.+=((grid_id, county_id, pointStartTime, pointEndTime))

              //切换下一个组
              headGrid = grid_id
              pointStartTime = startTime
              pointEndTime = endTime

            }
        }

        //处理最后一组
        resultPoint.+=((headGrid, lastCountyId, pointStartTime, pointEndTime))

        /**
         * mdn string comment '用户手机号码'
         * ,longi string comment '网格中心点经度'
         * ,lati string comment '网格中心点纬度'
         * ,grid_id string comment '停留点所在电信内部网格号'
         * ,county_id string comment '停留点区县'
         * ,duration string comment '机主在停留点停留的时间长度（分钟）,lTime-eTime'
         * ,grid_first_time string comment '网格第一个记录位置点时间（秒级）'
         * ,grid_last_time string comment '网格最后一个记录位置点时间（秒级）'
         *
         */

        //4、整理数据，这一块代码作为外面flatmap的返回值
        resultPoint.map {

          case (grid_id: String, county_id: String, pointStartTime: String, pointEndTime: String) =>

            //获取网格中心点经纬度
            val p: Point2D.Double = Grid.getCenter(grid_id.toLong)

            //经度
            val longi: String = p.getX.formatted("%.4f")
            //维度
            val lati: String = p.getY.formatted("%.4f")

            //计算停留时间（分钟）
            val format = new SimpleDateFormat("yyyyMMddHHmmss")
            val startTs: Long = format.parse(pointStartTime).getTime
            val endTs: Long = format.parse(pointEndTime).getTime
            val duration: String = ((endTs - startTs) / 60000).toString

            (mdn, longi, lati, grid_id, county_id, duration, pointStartTime, pointEndTime)

        }

    }

    //5、保存数据
    pointRDD
      .toDF()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_staypoint_msk_d/day_id=$day_id")

    //6、增加分区
    spark.sql(s"alter table dwi.dwi_staypoint_msk_d add if not exists partition(day_id='$day_id')")

  }
}
